package com.atguigu.flink.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/27 10:11
 */
public class Flink04_TableAPI_Connect_KafkaSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 通过 connect直接 将 外部系统（kafka） 抽象成 动态表,作为数据源

        Schema schema = new Schema()
                .field("a", DataTypes.STRING())
                .field("b", DataTypes.BIGINT())
                .field("c", DataTypes.INT());


        tableEnv
                .connect(
                        new Kafka()
                                .property("bootstrap.servers", "hadoop1:9092,hadoop2:9092,hadoop3:9092")
                                .property("group.id", "abc")
                                .version("universal")
                                .topic("flinksql0820")
                )
                .withFormat(new Csv().fieldDelimiter(',').lineDelimiter("\n"))
                .withSchema(schema)
                .createTemporaryTable("kafkaSourceTable");

        // 从 表名 获取 Table对象
        Table kafkaSourceTable = tableEnv.from("kafkaSourceTable");

        Table resultTable = kafkaSourceTable
                .groupBy($("a"))
                .select($("a"), $("c").sum().as("vcSum"));

        // 转成流、打印
        tableEnv.toRetractStream(resultTable, Row.class).print();


        env.execute();
    }
}
